package com.cloud.rou.im.client.handler;

import com.cloud.rou.im.client.api.ClientListener;
import com.cloud.rou.im.client.api.event.BindClientEvent;
import com.cloud.rou.im.client.api.event.ClientEventService;
import com.cloud.rou.im.client.api.event.DefaultClientListener;
import com.cloud.rou.im.client.api.event.IMClientEvent;
import com.cloud.rou.im.core.process.AbstractClientIMProcessor;
import com.cloud.rou.im.core.process.IMProcessor;
import com.cloud.rou.im.msg.AckMsg;
import com.cloud.rou.im.msg.ChatMsg;
import com.cloud.rou.im.msg.Msg;
import com.cloud.rou.im.msg.enumtype.AckType;
import com.cloud.rou.im.protocol.cmd.Cmd;
import com.cloud.rou.im.protocol.transfer.DataPacketFactory;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.lang.reflect.Method;
import java.time.Duration;
import java.util.Arrays;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;

/**
 * <p></p>
 *
 * @author GaoShuangWen
 */
public class BizHandler extends SimpleChannelInboundHandler<Cmd> implements GlobalHandler {

    private final IMProcessor<Msg> ackIMProcessor;

    private final IMProcessor<Msg> chatIMProcessor;

    private ChannelHandlerContext ctx;

    private final ClientEventService service;

    public BizHandler(ClientListener listener) {


        this.ackIMProcessor = new ClientAckIMProcessor();

        this.chatIMProcessor = new ClientChatIMProcessor();

        service = new ClientEventService();
    }

    public BizHandler() {
        this(new DefaultClientListener());
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.ctx = ctx;

    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Cmd cmd) throws Exception {
        Msg msg = cmd.getMsg();

        Optional<IMProcessor<Msg>> processor = getProcessor(msg);

        processor.ifPresent(msgIMProcessor -> msgIMProcessor.process(ctx, msg));

    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        service.offline();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        service.exception(cause);
    }

    @Override
    public void writeAndFlush(Msg msg) {
        ctx.writeAndFlush(DataPacketFactory.getPacket(msg));
        service.hasSend(msg);
    }


    class ClientAckIMProcessor extends AbstractClientIMProcessor<AckType, Msg> {

        @Override
        protected AckType getMapKey(Msg msg) {
            AckMsg ackMsg = (AckMsg) msg;
            return ackMsg.getAckType();
        }

        @Override
        public void registryProcessor() {
            // 注册ack回执
            Arrays.stream(AckType.values()).forEach(ackType ->
                    registry(ackType, msg -> service.methodInvokeByEvent(IMClientEvent.getEventByAckType(ackType),msg))
            );
        }
    }

    class ClientChatIMProcessor extends AbstractClientIMProcessor<Class<? extends Msg>, Msg> {
        @Override
        protected Class<? extends Msg> getMapKey(Msg msg) {
            return msg.getClass();
        }

        @Override
        public void registryProcessor() {
            registry(ChatMsg.class, msg -> service.read((ChatMsg) msg));
        }
    }

    private Optional<IMProcessor<Msg>> getProcessor(Msg msg) {
        if (msg instanceof ChatMsg) {
            return Optional.of(this.chatIMProcessor);
        }
        if (msg instanceof AckMsg) {
            return Optional.of(this.ackIMProcessor);
        }
        return Optional.empty();
    }
}
